package pub.cleangao.common.event;

import com.google.common.collect.HashBasedTable;
import com.google.common.eventbus.EventBus;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;


/**
 * 事件发布与监听
 */
@Slf4j
public class SimpleEventBus {


    // 处理方法存储
    private static final Map<String, Set<Consumer<IReactEvent>>> keyEventConsumer = new HashMap<>();
    private static final HashBasedTable<String, String, Set<Consumer<IReactEvent>>> nameEventConsumer = HashBasedTable.create();
    private static final Map<String, Set<Consumer<IReactEvent>>> typeEventConsumer = new HashMap<>();

    // 执行器
    private static final ExecutorService executor = Executors.newWorkStealingPool(Runtime.getRuntime().availableProcessors() * 2);

    private static final EventBus eventBus = new EventBus();

    /**
     * 广播事件
     */
    public static void publish(IReactEvent event) {
        if (event == null) {
            return;
        }
        react(event);
        eventBus.register(new ReactEventListener());
    }

    private static void react(IReactEvent event) {
        Set<Consumer<IReactEvent>> all = new HashSet<>();
        Set<Consumer<IReactEvent>> keyConsumers = keyEventConsumer.get(event.getKey());
        if (keyConsumers != null) {
            all.addAll(keyConsumers);
        }
        Set<Consumer<IReactEvent>> nameConsumers = nameEventConsumer.get(event.getModule(), event.getName());
        if (nameConsumers != null) {
            all.addAll(nameConsumers);
        }
        Set<Consumer<IReactEvent>> typeConsumers = typeEventConsumer.get(event.getClass().getName());
        if (typeConsumers != null) {
            all.addAll(typeConsumers);
        }
        if (all.isEmpty()) {
            return;
        }
        Iterator<Consumer<IReactEvent>> iterator = all.iterator();
        if (all.size() == 1) {
            // 单监听下使用调用者线程处理
            iterator.next().accept(event);
        }
        // 多监听,使用多线程处理
        while (iterator.hasNext()) {
            executor.submit(() -> {
                try {
                    iterator.next().accept(event);
                } catch (Exception e) {
                    log.error("ReactEvent process error: {}. event-key: {},event-module: {},event-name: {}", e.getMessage(), event.getKey(), event.getModule(), event.getName());
                }
            });
        }
    }

    /**
     * 注册对指定key的监听
     *
     * @param key      事件key
     * @param consumer 对事件的处理
     */
    public static void registerListenKey(String key, Consumer<IReactEvent> consumer) {
        Set<Consumer<IReactEvent>> consumerSet = keyEventConsumer.get(key);
        if (consumerSet == null) {
            consumerSet = new HashSet<>();
        }
        consumerSet.add(consumer);
        keyEventConsumer.put(key, consumerSet);
    }

    /**
     * 注册监听
     *
     * @param module   模块
     * @param name     事件名称
     * @param consumer 处理程序
     */
    public static void registerListenName(String module, String name, Consumer<IReactEvent> consumer) {
        Set<Consumer<IReactEvent>> consumerSet = nameEventConsumer.get(module, name);
        if (consumerSet == null) {
            consumerSet = new HashSet<>();
        }
        consumerSet.add(consumer);
        nameEventConsumer.put(module, name, consumerSet);
    }

    /**
     * 注册指定的实现类
     *
     * @param clazz    实现类类型
     * @param consumer 消费监听
     */
    public static void registerListenType(Class<? extends IReactEvent> clazz, Consumer<IReactEvent> consumer) {
        String clazzName = clazz.getName();
        Set<Consumer<IReactEvent>> consumerSet = typeEventConsumer.get(clazzName);
        if (consumerSet == null) {
            consumerSet = new HashSet<>();
        }
        consumerSet.add(consumer);
        typeEventConsumer.put(clazzName, consumerSet);
    }


}
